package com.flink.cus.wdopers;


import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;

public class stockTsWmarkPerdAssigner implements AssignerWithPeriodicWatermarks<Tuple3<Long, Float, Long>>{

    Long currentMaxTimestamp = 0L;
    Long maxOutOfOrderness = 100L;

    @Override
    public long extractTimestamp(Tuple3<Long, Float, Long> element, long previousElementTimestamp) {   
        currentMaxTimestamp = Math.max(element.f0, currentMaxTimestamp);
        return element.f0;
    }

    @Override
    public Watermark getCurrentWatermark() {
        return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
    }
    
}
